package cn.eversec.etl;

import com.eversec.sc.core.api.SpaceMeasure;
import com.eversec.sc.core.api.components.RowDataTransformer;
import com.eversec.sc.core.api.components.Updateable;
import com.eversec.sc.core.config.ComponentConfiguration;
import com.eversec.sc.core.exceptions.ComponentInitException;
import com.eversec.sc.core.types.RowData;
import org.apache.flink.configuration.Configuration;

import java.util.*;

/**
 * description
 *
 * @author 朱宗刚
 * @version 1.0.0
 * @date 2021/7/19
 */

public class DemoTransform extends RowDataTransformer implements SpaceMeasure, Updateable {
    private Map<String, Long> thresholds;
    private Set<String> thresholdKey = new HashSet<>(3);

    @Override
    public int measure() {
        return 0;
    }

    @Override
    public RowData doTransform(RowData rowData) throws Exception {
        return null;
    }

    @Override
    public void onUpdate(List<ComponentConfiguration.Binding> list) {
        thresholds = getThreshold(list.get(0).getBindDatas());
    }

    @Override
    public void onInit(Configuration globalConfiguration, ComponentConfiguration configuration) throws ComponentInitException {
        super.onInit(globalConfiguration, configuration);
        thresholdKey.addAll(Arrays.asList("a", "b", "c"));
        thresholds = getThreshold(configuration.getBindings().get(0).getBindDatas());
    }

    private Map<String, Long> getThreshold(List<Map<String, String>> threshold) {
        Map<String, Long> ds = new HashMap<>(3);
        threshold.forEach(item -> {
            ds.put(item.get("arg_code"), Optional.of(Long.parseLong(item.get("arg_value"))).orElse(10L));
        });

        return ds;
    }
}
